package com.young.trigger.handler;

import com.google.gson.Gson;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import com.young.commom.httpclient.HttpClientTemplate;
import com.young.commom.httpclient.HttpResult;
import com.young.common.InetTool;
import com.young.common.Utils;
import com.young.core.domain.JobDO;
import com.young.core.domain.NodeDO;
import com.young.core.domain.TraceDO;
import com.young.core.domain.TriggerDO;
import com.young.core.enums.HttpStatusEnum;
import com.young.core.enums.JobStrategyEnum;
import com.young.core.manager.JobManager;
import com.young.core.manager.NodeManager;
import com.young.core.manager.TraceManager;
import com.young.core.manager.TriggerManager;
import com.young.trigger.configuration.WebProperties;
import com.young.trigger.event.CallFailedEvent;
import com.young.trigger.event.EmptyNodeEvent;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.io.Closeable;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;

/**
 * @description：
 * @author: yangyahui01
 * @date: 11/28/24 5:24 PM
 */
@Component
public class SlaveHandler implements Closeable {

    private static final Logger LOGGER = LoggerFactory.getLogger(SlaveHandler.class);

    private static final int CET = 128;

    private final Random random = new Random();

    private final ExecutorService te = Executors.newFixedThreadPool(CET, new NamedThreadFactory("Trigger - "));

    private final ExecutorService ce = Executors.newFixedThreadPool(CET, new NamedThreadFactory("HttpCaller - "));

    private final DelayQueue<HttpRequest> queue = new DelayQueue<>();

    @Autowired
    private GrpcClient grpcClient;

    @Autowired
    private TriggerManager triggerManager;

    @Autowired
    private JobManager jobManager;

    @Autowired
    private NodeManager nodeManager;

    @Autowired
    private TraceManager traceManager;

    @Autowired
    private WebProperties webProperties;

    @Autowired
    private ApplicationEventPublisher applicationEventPublisher;

    @Autowired
    private HttpClientTemplate httpClientTemplate;


    public static class NamedThreadFactory implements ThreadFactory {
        private volatile int sequence = 0;
        private final boolean daemon;
        private final String name;

        NamedThreadFactory(String name) {
            this.name = name;
            this.daemon = true;
        }

        NamedThreadFactory(String name, boolean daemon) {
            this.name = name;
            this.daemon = daemon;
        }

        public Thread newThread(Runnable runnable) {
            Thread thread = new Thread(runnable);
            thread.setDaemon(this.daemon);
            thread.setName(this.name + this.sequence++);
            return thread;
        }
    }

    public SlaveHandler() {
        for (int i = 0; i < CET; i++) {
            ce.execute(httpRunner());
        }
        LOGGER.info("Slave - http线程池任务已启动就绪,{}", CET);
    }

    /**
     * 从数据库中查询分配的任务，trigger并立即执行
     */
    @Scheduled(initialDelay = 5000, fixedDelay = 200)
    public synchronized void fetch() {
        //如果已经关闭，或者进入了local模式，则无需再进行db获取新的trigger
        if (BrokerContext.isRetired()) {
            return;
        }

        try {
            Collection<TriggerDO> triggers = triggerManager.getAll(InetTool.LOCAL_IP);

            List<Integer> ids = new ArrayList<>();

            for (TriggerDO trigger : triggers) {
                if (BrokerContext.isRetired()) {
                    break;
                }
                te.execute(triggerRunner(trigger));
                ids.add(trigger.getId());
            }

            triggerManager.delete(ids);
        } catch (Exception e) {
            LOGGER.error("Slave - 任务分发执行失败", e);
        }

    }


    private Runnable triggerRunner(final TriggerDO trigger) {

        return new Runnable() {
            @Override
            public void run() {
                try {
                    List<HttpRequest> requests = convert(trigger);
                    //如果节点列表不为空，则调度，否则提示异常
                    if (!requests.isEmpty()) {
                        requests.forEach(request -> {
                            //添加到队列
                            queue.offer(request);
                        });
                    } else {
                        LOGGER.info("Slave - 客户端节点为空,uniqueCode - [" + trigger.getJob().getUniqueCode() + "]");
                        applicationEventPublisher.publishEvent(new EmptyNodeEvent(trigger.getJob(), InetTool.LOCAL_IP));
                    }
                } catch (ExecutionException e) {
                    LOGGER.error("Slave - trigger调度失败,project - {},jobId - {}", trigger.getProject(), trigger.getJobId(), e);
                }
            }
        };
    }

    private List<HttpRequest> convert(TriggerDO trigger) throws ExecutionException {
        //查询任务
        JobDO job = jobManager.getCache(trigger.getJobId());

        List<NodeDO> candidates = this.getNodes(job);

        trigger.setJob(job);

        //如果节点列表不为空，则调度，否则提示异常
        if (candidates.isEmpty()) {
            return new ArrayList<>();
        }


        JobStrategyEnum strategy = JobStrategyEnum.codeOf(job.getStrategy());


        List<HttpRequest> requests = choose(strategy, candidates);


        String traceId = trace(job, trigger, requests.size());

        //填充核心值
        for (HttpRequest request : requests) {
            request.job = job;
            request.traceId = traceId;
            request.ts = System.currentTimeMillis();
            if (trigger.getTs() != null) {
                request.ts = trigger.getTs();
            }
        }
        return requests;
    }

    /**
     * 构建任务执行记录
     *
     * @param job
     * @param trigger
     * @param total
     * @return
     */
    private String trace(JobDO job, TriggerDO trigger, int total) {
        TraceDO trace = new TraceDO();
        trace.setProject(job.getProject());
        trace.setJobId(job.getId());
        trace.setName(job.getName());
        trace.setClazz(job.getClazz());
        trace.setMethod(job.getMethod());
        trace.setUniqueCode(job.getUniqueCode());
        String traceId = traceId(job.getId(), trigger.getId());
        trace.setTraceId(traceId);
        trace.setTriggerNode(InetTool.LOCAL_IP);
        trace.setTotal(total);
        trace.setData(job.getData());
        traceManager.insert(trace);

        return traceId;
    }

    /* 生成 traceId */
    private String traceId(int jobId, int triggerId) {
        long tid = (triggerId <= 0 ? System.currentTimeMillis() : triggerId);
        //保证长度，相对一致
        return StringUtils.leftPad(jobId + "." + tid + "." + random.nextInt(6), 24, "0");
    }

    /**
     * 根据调度策略选择业务端 node 节点
     *
     * @param strategy
     * @param candidates
     * @return
     */
    private List<HttpRequest> choose(JobStrategyEnum strategy, List<NodeDO> candidates) {
        List<HttpRequest> requests = new ArrayList<>();
        int size = candidates.size();

        if (strategy == JobStrategyEnum.ALL || strategy == JobStrategyEnum.SHARDING) {
            for (int i = 0; i < size; i++) {
                NodeDO node = candidates.get(i);
                HttpRequest request = new HttpRequest();
                request.index = i + 1;
                request.total = size;
                request.candidateNodes = Collections.singletonList(node);
                requests.add(request);
            }
        } else if (strategy == JobStrategyEnum.FIRST) {
            HttpRequest request = new HttpRequest();
            request.candidateNodes = Arrays.asList(candidates.get(0));
            requests.add(request);
        } else {
            Collections.shuffle(candidates);//随机一次
            HttpRequest request = new HttpRequest();
            request.candidateNodes = candidates;
            requests.add(request);
        }

        return requests;
    }


    /**
     * 得到该任务所在业务端的存活且同 group 节点
     *
     * @param job
     * @return
     * @throws ExecutionException
     */
    private List<NodeDO> getNodes(JobDO job) throws ExecutionException {
        List<NodeDO> nodes = new ArrayList<>(nodeManager.getAliveCache(job.getProject(), 90));

        Set<String> ips = nodes.stream().map(node -> node.getIp()).collect(Collectors.toSet());
        String group = job.getGroup();
        //对于自定义节点
        if (group.isEmpty()) {
            String[] segments = job.getCustomNodes().split("\\s+");
            return Utils.wrap(segments, true, (segment) -> {
                NodeDO node = new NodeDO(segment, job.getPort());
                if (!ips.contains(segment)) {
                    node.setHttpProtocol(false);
                }
                return node;
            });
        }

        Set<String> segments = Utils.array2Set(group.split(","), true);
        nodes.removeIf((node) -> !segments.contains(node.getGroup()));
        return nodes;

    }

    private Runnable httpRunner() {
        return new Runnable() {
            @Override
            public void run() {
                while (!Thread.interrupted()) {
                    try {
                        HttpRequest request = queue.take();

                        try {
                            NodeDO node = request.selected();

                            int code = 0;

                            if (node.isHttpProtocol()) {
                                //http 请求业务端
                                code = httpCall(request);
                            } else {
                                //grpc 请求业务端
                                code = grpcClient.call(request);
                            }

                            LOGGER.info("Slave - 调度结果, traceId - {},{}/{}.{} code:{}", request.traceId, request.job.getProject(), request.job.getClazz(), request.job.getMethod(), code);

                            record(request, HttpStatusEnum.uf(code).standard);

                            if (!HttpStatusEnum.isUSuccess(code)) {
                                int retryTimes = request.getRetryTimes();//
                                if (retryTimes < webProperties.getRetryTimes()) {
                                    request.retry(retryTimes + 1, TimeUnit.SECONDS);
                                    queue.offer(request);
                                    LOGGER.debug("Slave - http调度失败即将重试,{}", request.traceId);
                                } else {
                                    LOGGER.info("Slave - http调度失败,traceId - {},{}/{}.{}",
                                            request.traceId,
                                            request.job.getProject(),
                                            request.job.getClazz(),
                                            request.job.getMethod());
                                    applicationEventPublisher.publishEvent(new CallFailedEvent(request.job, node.getIp() + ":" + node.getPort()));
                                }
                            }
                        } catch (Exception e) {
                            LOGGER.error("Slave - 调用线程池出错", e);
                        }
                    } catch (InterruptedException e) {
                        LOGGER.error("Slave - http调用线程池退出");
                    }
                }
            }
        };
    }

    private void record(HttpRequest request, int status) {
        String ip = request.selected().getIp();
        int port = request.selected().getPort();
        try {
            TraceDO.NodeTrace nodeTrace = new TraceDO.NodeTrace();
            nodeTrace.setIp(ip);
            nodeTrace.setPort(port);
            nodeTrace.setIndex(request.index);
            nodeTrace.setRetryNumber(request.retryTimes);
            nodeTrace.setStatus(status);
            nodeTrace.setTime(new Date());
            traceManager.append(request.traceId, nodeTrace);
        } catch (Exception e) {
            LOGGER.error("Slave - 记录任务trace失败,traceId - {},{}/{}.{},node - [{}:{}]", request.traceId,
                    request.job.getProject(),
                    request.job.getClazz(),
                    request.job.getMethod(),
                    ip, port, e);
        }
    }

    /**
     * 业务端地址
     */
    private static final String CLIENT_URL_OLD = "http://%1$s:%2$s/young/schedule/client/call";

    /**
     * 请求业务端
     *
     * @param request
     */
    private int httpCall(HttpRequest request) {


        String ip = request.selected().getIp();

        int port = request.selected().getPort();

        try {
            String url = String.format(CLIENT_URL_OLD, ip, port);

            String content = new Gson().toJson(httpParameters(request));

            HttpResult result = httpClientTemplate.post(url, content);

            if (!result.isSuccess()) {
                LOGGER.info("Slave - http请求网络失败,traceId - {},code - {}", request.traceId, result.getCode());
                return result.getCode();
            }

            JsonObject object = JsonParser.parseString(result.getContent()).getAsJsonObject();

            if (object.has("message")) {
                String message = object.get("message").getAsString();
                LOGGER.info("Slave - 任务回调信息,traceId:{},message:{}", request.traceId, message);
            }

            if (object.has("status")) {
                return object.get("status").getAsInt();
            }

        } catch (Exception e) {
            LOGGER.error("Slave - http请求失败,traceId - {},jobId - {},{}/{}.{},client:[{}:{}]",
                    request.traceId,
                    request.job.getId(),
                    request.job.getProject(),
                    request.job.getClazz(),
                    request.job.getMethod(),
                    ip, port, e);
        }
        return -2;
    }

    private Object httpParameters(HttpRequest httpRequest) {
        Map<String, Object> parameters = new HashMap<>();
        parameters.put("traceId", httpRequest.traceId);
        parameters.put("data", httpRequest.job.getData());
        parameters.put("clazz", httpRequest.job.getClazz());
        parameters.put("method", httpRequest.job.getMethod());
        parameters.put("strategy", httpRequest.job.getStrategy());
        parameters.put("index", httpRequest.index);
        parameters.put("total", httpRequest.total);
        parameters.put("uniqueCode", httpRequest.job.getUniqueCode());

        //历史兼容
        parameters.put("jobStrategyCode", httpRequest.job.getStrategy());
        parameters.put("jobNodeNum", httpRequest.total);
        parameters.put("jobNodeIndex", httpRequest.index);
        parameters.put("jobUniqCode", httpRequest.job.getUniqueCode());

        List<Map<String, Object>> nodes = new ArrayList<>();
        Map<String, Object> mapper = new HashMap<>();
        mapper.put("ip", webProperties.getDomain() == null ? InetTool.getLocalIp() : webProperties.getDomain());
        mapper.put("port", webProperties.getPort() == null ? 80 : webProperties.getPort());
        nodes.add(mapper);
        parameters.put("callBackNodes", nodes);
        return parameters;
    }


    @Override
    public void close() throws IOException {

        te.shutdown();
        LOGGER.info("Slave - trigger线程池准备退出");
        try {
            te.awaitTermination(3, TimeUnit.SECONDS);
        } catch (Exception e) {
            //
        }

        for (int i = 0; i < 12; i++) {
            int size = queue.size();
            LOGGER.info("Slave - http线程池准备退出,剩余任务:" + size);
            if (size != 0) {
                try {
                    Thread.sleep(200);
                } catch (InterruptedException e) {
                    //
                }
            } else {
                break;
            }
        }

        ce.shutdown();
        try {
            ce.awaitTermination(3, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            //
        }

        queue.clear();
        LOGGER.info("Slave - handler已完全退出");

    }
}
